on the performance thing, i suppose it's possible we could do pause tracking using the internal state from #129, and move the #fetch() logic that's current in MessagesStream up into Consumer. If we did that, then each MessagesStream instance could flip itself between paused/resumed when it fills up and drains, and we'd go back to having one consolidated fetch across streams. that's a more invasive refactor though.

I'm not exactly convinced about the DX of onAssignmentChange and what happens to the current streams. I think a better approach would be to terminate/destroy all streams and have the users re-create them.

WDYT?

def agree the current API is a little wonky, couple thoughts:

• for streams where the consuming code is quick to spin up/down, and buffering is minimal (e.g. default of 1024), tearing down all the streams and recreating would be conceptually simpler without much downside. but i do worry a bit about the perf overhead of doing that for every rebalance in low latency systems where spinning up work for a partition takes significant time or the consumer loses a ton of buffered messages and needs to refetch all of them. for example in our indexing service we buffer up to ~100k messages per partition and do work with them in large batches, with a goal of well under 1s of end to end latency. i don't know exactly how long it would take to do a rebalance and then fully tear down and re-hydrate all of that state and buffer content — could be not too bad or could be seconds. goal of keeping existing streams alive was to create minimal spinup/teardown disruption on rebalance for perf. maybe we can just go with the destroy-and-recreate behavior initially since it's simpler, and if i see long rebalance times in production monitoring we can come up with a higher perf option.

• if we did close all streams and re-create, what are you thinking of in terms of DX for picking up those new streams after a rebalance? are you expecting someone would re-call consumeByPartition after somehow detecting a rebalance or something different?

an additional issue i was just thinking about this morning was the MANUAL offset mode, where devs pass in offsets for partitions themselves... even in today's regular consume() api this actually doesn't really work with rebalances if app code is tracking offsets manually in some other system like redis — if another consumer owns a partition initially and then it gets reassigned to you after you already have a consume() going, it will use an old offset you probably don't want. and there is currently now equivalent to the seek() api in kafkajs that lets you move to a different offset once consumption has started. i'll think a bit more about that maybe we just need a seek() method and that problem goes away. i'll open a separate issue to figure that out. maybe if we have people re-calling consumeByPartition() on every rebalance they'd be able to supply new offsets every time as a way to resolve.

About the DX I agree with @mcollina: let's destroy everything and create from scratch.
As a good DX I would use another type of stream itself so people have several ways to consume them.

Something like:

// The returning stream is in objectMode.
const perPartitionStreams = consumer.consumeByPartition({ topics: ['foo'] })

// A new one is 
for await (const streams of perPartitionStreams) {
  for(const {topic, partition, stream} of streams) {
    // Do your magic. The stream will automatically close when assignments change.
  }
}

Note that batching shouldn't be an issue as in-memory messages are still processed when the stream is gracefully closed.

About the MANUAL problem I agree with you and the seek method is probably a good idea, just pick a better and explanatory name like setOffsetsToFetch or something like that.